import KafkaStream.{dbPassword, dbUser, getConnection, jdbcUrl}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

import java.sql.{Connection, DriverManager, PreparedStatement, SQLException}
object RegistStream {
  private val jdbcUrl = "jdbc:mysql://localhost:3306/stats?serverTimezone=Asia/Shanghai"
  private val dbUser = "root"
  private val dbPassword = "123456"
  def getConnection(): Connection = {
    DriverManager.getConnection(jdbcUrl, dbUser, dbPassword)
  }

  def main(args: Array[String]): Unit = {
    // Spark 配置
    val conf = new SparkConf().setMaster("local[*]").setAppName("KafkaSparkStream1")
    val ssc = new StreamingContext(conf, Seconds(2))

    ssc.sparkContext.setLogLevel("error")
    // Kafka 配置
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.244.11:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "hael4",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("lol")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val genderCounts = stream.map(record => record.value.split("\t"))
      .filter(_.length == 7) // 确保我们只处理正确的记录
      .map(fields => ((fields(2).toInt, fields(6)), 1)) // (性别, 在籍状态) -> 计数
      .reduceByKey(_ + _)

    genderCounts.foreachRDD { rdd =>
      val result = rdd.collect()
      result.foreach { case ((gender, enrollmentStatus), count) =>
        println(s"Gender: ${if (gender == 1) "Male" else "Female"}, Status: $enrollmentStatus -> Count: $count")
        saveToMySQL(gender.toString,enrollmentStatus, count)
      }
    }

    // 启动 Spark Streaming
    ssc.start()
    ssc.awaitTermination()
  }

  def saveToMySQL(gender: String, status: String, count: Int): Unit = {
    // Class.forName("com.mysql.cj.jdbc.Driver")
    //    val url = "jdbc:mysql://localhost:3306/your_database"
    //    val user = "your_username"
    //    val password = "your_password"

    val conn = getConnection()
    try {
      val sql = "  INSERT INTO regist (gender, status, count) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE count = count + VALUES(count)"
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, gender)
      stmt.setString(2, status)
      stmt.setInt(3, count)
      stmt.executeUpdate()
    } catch {
      case e: SQLException => e.printStackTrace()
    } finally {
      if (conn != null) conn.close()
    }


  }

}